import pymysql
import pandas as pd
import asyncio
import time
import datetime
import traceback
import os
import sys
import numpy as np
import pytz
# print(os.path.join(os.path.split(os.path.split(sys.path[0])[0])[0], "bin"))
# # 添加导包路径
# sys.path.append(os.path.join(os.path.split(os.path.split(sys.path[0])[0])[0], "bin"))
sys.path.append(r"D:\my_work\ac_lingxing_api\ac_lingxing_api")
from multiprocessing import  Process, Queue, Pool
from bin.openapi import OpenApiBase
from sqlalchemy import create_engine
from itertools import product
from itertools import islice



class tools:
    
    @staticmethod
    def chunked_iterable(iterable, size):
        it = iter(iterable)
        return iter(lambda: tuple(islice(it, size)), ())
    
    
    @staticmethod
    def get_engine():
        engine = create_engine('mysql+pymysql://user_rpa:123456@192.168.8.10:3306/rpa?charset=utf8mb4')
        return engine
    
    @staticmethod
    def get_app_token():
        engine = tools.get_engine()
        res = pd.read_sql("select appId, appSecret from lx_app_token", con=engine)
        app_id_lst = []
        app_secret_lst = []
        for i, data in res.iterrows():
            app_id_lst.append(data['appId'])
            app_secret_lst.append(data['appSecret'])
        return app_id_lst, app_secret_lst

    @staticmethod
    def get_email():
        engine = tools.get_engine()
        res = pd.read_sql("select distinct store_username from amz_store", con=engine)
        email_lst = list(res['store_username'])
        print(email_lst)
        return email_lst
    
    @staticmethod
    def get_date_lst(days=1):
        today = datetime.datetime.now().strftime("%Y-%m-%d")
        date_lst = []
        for i in range(0, days+1):
            date_lst.append((datetime.datetime.now()+ datetime.timedelta(-i)).strftime("%Y-%m-%d"))
        print(date_lst)
        return date_lst
    
    @staticmethod
    def get_date_lst2(start_date:int, end_date:int):
        date_today = datetime.datetime.now()+ datetime.timedelta(days=start_date)
        end_date = date_today + datetime.timedelta(days=end_date)
        date_lst = []
        while date_today != end_date:
            date_lst.append(date_today.strftime("%Y-%m-%d"))
            date_today = date_today+datetime.timedelta(days=-1)
        return date_lst
        


class lx_api:

    def __init__(self, app_id, app_secret, email_lst):
        
        self.op_api = OpenApiBase("https://openapi.lingxing.com", app_id, app_secret)
        self.access_token = ""
        self.refresh_token = ""
        self.engine = tools.get_engine()
        self.table_name = "lx_cus_service22"
        self.url = "/erp/sc/data/mail/lists"
        self.email_lst = email_lst
    
    # def data_exists(self, webmail_uuid):
    #     connection = pymysql.connect(user='user_rpa', password='123456', host='192.168.8.10', database='ad_api')
    #     cur = connection.cursor()
    #     query = f"select * from {self.table_name} where webmail_uuid = '{webmail_uuid}'"
    #     cur.execute(query)
    #     result = cur.fetchone()
    #     cur.close()
    #     connection.close()
    #     return result is not None

        

    async def generate_access_token(self):
        resp = await self.op_api.generate_access_token()
        self.access_token = resp.access_token
        self.refresh_token = resp.refresh_token
        assert resp.access_token
        # return resp.dict()
    
    async def refresh_token_handler(self):
        resp = await self.op_api.refresh_token(self.refresh_token)
        self.refresh_token =  resp.refresh_token
        self.access_token = resp.access_token
        assert resp.access_token

    async def check_token(self, data):
        if data['code'] == 0:
            return True
        elif data['code'] in [2001005, 2001003, 2001008, 2001009]:
            await self.generate_access_token()
            return False
        else:
            print(data['code'])
            return False
    
    async def get_email_data(self, email, date, flag):
        req_body = {
            "flag":flag,
            "email":email,
            "start_date":date,
            "end_date":date,
            "offset":0,
            "length":20
        }
        res_lst = []
        while 1:
            try:
                res = await self.op_api.request(self.access_token, self.url, "POST", req_body=req_body)
            except:
                print("异常:", traceback.format_exc())
                time.sleep(1)
                continue
            resp = res.dict()
            if not await self.check_token(resp):
                continue
            if len(resp['data']) != 0:
                print(resp['data'])
                res_lst = res_lst + resp['data']
                req_body['offset'] = req_body['offset'] + 20
            else:
                break
        if len(res_lst) == 0:
            print("无数据")
            return []
        res_df = pd.DataFrame(res_lst)
        res_df['update_time'] = pd.to_datetime(datetime.datetime.now())
        
        if len(res_df)>0:
            existing_ids = pd.read_sql_query(f"SELECT webmail_uuid FROM {self.table_name} where DATE(date) = date", con=self.engine)
            existing_ids_list = existing_ids['webmail_uuid'].tolist()

            res_df = res_df[~res_df['webmail_uuid'].isin(existing_ids_list)]
            res_df.to_sql(self.table_name, con=self.engine, if_exists="append", index=False)

            print(f"{date} 日期 {email} 数据已插入")
        else:
            print("无数据")
    async def hanlder(self, semaphore, email, date, flag):
        async with semaphore:
            await self.get_email_data(email, date, flag)
            
        
    async def main(self):
        semaphore = asyncio.Semaphore(1)
        await self.generate_access_token()
        tasks = []
        for data in self.email_lst:
            email, date = data
            tasks.append(asyncio.create_task(self.hanlder(semaphore, email, date, "sent")))
            tasks.append(asyncio.create_task(self.hanlder(semaphore, email, date, "receive")))
        result = await asyncio.gather(*tasks)
            
        
        

def run(app_id, app_secret, email_lst):
    
    lx_app = lx_api(app_id, app_secret, email_lst)
    
    asyncio.run(lx_app.main())
    
        

def main():
    app_id_lst, app_secret_lst = tools.get_app_token()
    email_lat = tools.get_email()
    date_lst = tools.get_date_lst(1)
    result = list(product(email_lat, date_lst))
    
    result_lst = np.array_split(result, 8)
    pool = Pool(8)
    for i in range(8):
        print(app_id_lst[i], app_secret_lst[i])
        pool.apply_async(func=run, args=(app_id_lst[i], app_secret_lst[i], result_lst[i]))
    pool.close()
    pool.join()
if __name__ == "__main__":  
    main()